package org.cep

import org.FlinkStreamApp
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

import scala.collection.Map

/**
 * description ：cep example
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/24 20:49
 * modified By ：
 * version:    : 1.0
 */
object FlinkCEPExample extends FlinkStreamApp {

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 获取源数据
    val source = environment.fromElements(
      LoginEvent("1", "192.168.2.101", "fail", "1"),
      LoginEvent("1", "192.168.2.102", "fail", "2"),
      LoginEvent("1", "192.168.2.103", "fail", "3"),
      LoginEvent("2", "192.168.2.104", "fail", "4")
    )

    // 处理
    val stream = source
      .assignAscendingTimestamps(_.eventTime.toLong * 1000)
      .keyBy(_.userId)

    // cep
    val pattern = Pattern
      .begin[LoginEvent]("first").where(_.eventType == "fail")
      .next("second").where(_.eventType == "fail")
      .next("third").where(_.eventType == "fail")
      .within(Time.seconds(10))

    // cep 匹配
    val patternStream = CEP.pattern(stream, pattern)

    // 过滤 cep 匹配的结果
    val patternStreamResult = patternStream
      .select((p: Map[String, Iterable[LoginEvent]]) => {
        val first = p.getOrElse("first", null).toIterator.next()
        val second = p.getOrElse("second", null).toIterator.next()
        val third = p.getOrElse("third", null).toIterator.next()
        s"用户id = ${first.userId}, 在 10s 内连续登录失败了3次， ip 地址分别是 ${first.ip}, ${second.ip}, ${third.ip}"
      })

    // 输出 cep 匹配的结果
    patternStreamResult.print
  }

  case class LoginEvent(userId: String, ip: String, eventType: String, eventTime: String)

}
